package com.ibm.cps.processors;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;

import org.apache.log4j.Logger;

import com.google.gson.Gson;
import com.ibm.cps.dft.DFTOperator;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.cps.message.ParsedDataMessageBuilder;
import com.ibm.cps.sql.expression.function.meta.FieldNameParam;
import com.ibm.cps.sql.expression.function.meta.IgnoredInFunctionCall;
import com.ibm.cps.sql.expression.function.meta.ValueParam;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.util.exception.CPSException;
// need to extends
//TODO: test out the serialization / deserialization
@Info(Name=MessageFields.BATCH_DFT_TYPE,Type=Info.ProcessorType.Vector)
public class BasicDFTProcessor extends SchemaInheritiveProcessor {

	@FieldNameParam
	private String inputField;
	@ValueParam
	private int coefficient;
	@IgnoredInFunctionCall@FieldNameParam
	private String outputField;

	// private HashMap<Long, IncDFTOperator> tsdft_operators;
	private HashMap<Comparable, DFTOperator> tsdft_operators;
	private static final Logger logger = Logger.getLogger(BasicDFTProcessor.class);;

	public BasicDFTProcessor(String tenantId, String processorId, String source, ITimeSeriesSchema parentSchema) throws CPSException {
		super(tenantId, processorId, source, parentSchema);
		this.parentSchema = parentSchema;
		logger.info("parent schema for dft is " + parentSchema.getSchemaID());
		this.tsdft_operators = new HashMap<>();
	}


	@Override
	public void dispose() {
	}

	@Override
	public String toString() {
		Gson gson = new Gson();
		return gson.toJson(this);
	}


	@Override
	protected List<NewField> defineFieldsToBeAdded() {
		NewField f = new NewField(outputField, "double[]");
		List<NewField> fields= new ArrayList<>();
		fields.add(f);
		return fields;
	}


	@Override
	protected Collection<ParsedDataMessage> runProcessorLogic(ITimeSeriesSchema parentSchema,
			ITimeSeriesSchema outputSchema, ParsedDataMessage message) throws CPSException {
		Collection<ParsedDataMessage> messages = new ArrayList<>();
		ParsedDataMessageBuilder outputBuilder = ParsedDataMessageBuilder.createInhertiveBuilder(outputSchema, message, parentSchema);
		Comparable tskey = message.getTsKey();
		DFTOperator tsdftoper = null;

		if (tsdft_operators.containsKey(tskey)) {
			tsdftoper = tsdft_operators.get(tskey);
		} else {
			tsdftoper = new DFTOperator();
			tsdft_operators.put(tskey, tsdftoper);
		}

		int index = parentSchema.getTagId(inputField);
		Double[] datas =  message.getArray(index);
		
		Double[] stepdata = new Double[datas.length];
		for (int i = 0; i < stepdata.length; i++) {
			stepdata[i] = (Double) datas[i];
		}
		

		
		Double[] dftArr = tsdftoper.ComputDFTModulusVecotr(stepdata);
		System.out.println("==============BasicDFT ============");
		System.out.println("dftoperator output: " + Arrays.toString(dftArr));
		if (dftArr != null) {
			int size = Math.min(coefficient, dftArr.length);
			Double[] result = new Double[size];
			for (int i = 0; i < result.length; i++) {
				result[i] = dftArr[i];
			}
			outputBuilder.append(outputField, result);
		}


		messages.add(outputBuilder.toParsedDataMessage());
		return messages;
	}


	@Override
	public void initialize() throws CPSException {
		
	}



}
